package com.fulin.thread.condition;

import java.util.HashSet;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

public class TestBlockqueue {

    private static LinkedBlockingQueue<Integer> pool = new LinkedBlockingQueue<Integer>(1024);

    private static volatile boolean stop = false;

    public static void main(String[] args) throws InterruptedException {
        Thread p1 = new Thread(()->{
            for(int i = 1;i <= 10000000;i++){
                put(i);
            }
        },"p1");

        Thread p2 = new Thread(()->{
            for(int i = 10000001;i <= 20000000;i++){
                put(i);
            }
        },"p2");


        Thread m1 = new Thread(new Consumer(),"m1");

        Thread m2 = new Thread(new Consumer(),"m2");

        p1.start();
        p2.start();
        m1.start();
        m2.start();


        Thread.sleep(4000L);
        stop = true;
        while(pool.size() != 0){
            Thread.sleep(200);
        }
        System.out.println("阻塞队列中已经清空了");
        Consumer.refsh = true;
        if(!m1.getState().equals(Thread.State.WAITING)){
            Thread.sleep(200);
        }
        m1.interrupt();
        if(!m2.getState().equals(Thread.State.WAITING)){
            Thread.sleep(200);
        }
        m2.interrupt();
        Thread.sleep(2000);
    }

    private static Object lock = new Object();

    private static void put(Integer ele){
        try {
            if(stop){
                System.out.println("开始暂停,生产者等待");
                synchronized (lock){
                    lock.wait();
                }

            }
            pool.put(ele);
        } catch (InterruptedException e) {
            System.out.println("生产者被打断了");
        }
    }

    private static Integer take(){
        Integer take = null;
        try {
            take = pool.take();
        } catch (InterruptedException e) {
            System.out.println("消费者被打断了");
        }
        return take;
    }

    private static class Consumer implements Runnable{

        public static volatile boolean refsh = false;

        @Override
        public void run() {
            String name = Thread.currentThread().getName();
            Set<Integer> holder = ThreadHolder.get();
            while(true){
                if(holder.size() >= 60){
                    try {
                        System.out.println(name + " : 临时缓存满了,开始处理");
                        Thread.sleep(1000);
                        holder.clear();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                Integer take = take();
                if(take != null){
                    ThreadHolder.add(take);
                }else{
                    if(!refsh){
                        return;
                    }

                    System.out.println(name + " 开始强制处理临时缓存中的数据");
                    System.out.println(name + " 临时缓存中有" + holder.size() + " 条数据");
                }
            }
        }
    }

    private static class ThreadHolder{
        private static ThreadLocal<Set<Integer>> holder = new ThreadLocal<Set<Integer>>() {
            @Override
            protected HashSet<Integer> initialValue() {
                return new HashSet<Integer>();
            }
        };

        public static void add(Integer ele){
            holder.get().add(ele);
        }

        public static Set<Integer> get(){
            return holder.get();
        }
    }
}
